package DivorceTopOne4;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class DivorceTopOneDriver {

    public static void main(String[] args) throws Exception {

            //配置文件对象
            Configuration conf = new Configuration();
            // 创建作业实例
            Job job = Job.getInstance(conf, DivorceTopOneDriver.class.getSimpleName());
            // 设置作业驱动类
            job.setJarByClass(DivorceTopOneDriver.class);

            // 设置作业mapper reducer类
            job.setMapperClass(DivorceTopOneMapper.class);
            job.setReducerClass(DivorceTopOneReducer.class);

            // 设置作业mapper阶段输出key value数据类型NullWritable
            job.setMapOutputKeyClass(DivorceBean.class);
            job.setMapOutputValueClass(NullWritable.class);

            //设置作业reducer阶段输出key value数据类型 也就是程序最终输出数据类型
            job.setOutputKeyClass(DivorceBean.class);
            job.setOutputValueClass(NullWritable.class);

            //todo 设置自定义分组
            job.setGroupingComparatorClass(DivorceGroupComparator.class);

            // 配置作业的输入数据路径
            FileInputFormat.addInputPath(job, new Path("DataOut/divorceSum4"));
            // 配置作业的输出数据路径
            FileOutputFormat.setOutputPath(job, new Path("DataOut/divorceTopOne4"));
            //判断输出路径是否存在 如果存在删除
            FileSystem fs = FileSystem.get(conf);
            if (fs.exists(new Path("DataOut/divorceTopOne4"))) {
                fs.delete(new Path("DataOut/divorceTopOne4"), true);
            }
            // 提交作业并等待执行完成
            boolean resultFlag = job.waitForCompletion(true);
            //程序退出
            System.exit(resultFlag ? 0 : 1);
    }
}
